package com.dongnao.network.nio.reactor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Date;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;

/**
 * 工作线程池版的Reactor模式
 */
public class WorkerThreadPoolReactor {

    public static void main(String[] args) {
        try {
            Reactor reactor = new Reactor(8080);
            reactor.run();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 处理多个客户端并发请求
     */
    private static class Reactor implements Runnable {
        final Selector sel;
        final ServerSocketChannel serverSocket;

        public Reactor(int port) throws IOException {
            sel = Selector.open();
            serverSocket = ServerSocketChannel.open();
            serverSocket.configureBlocking(false);
            //获取serverSocket在Selector中注册的标识sk，sk绑定了serverSocket与Selector的关系，维护channel事件
            SelectionKey sk = serverSocket.register(sel, SelectionKey.OP_ACCEPT);
            serverSocket.socket().bind(new InetSocketAddress(port));
//            serverSocket.bind(new InetSocketAddress(port));
            //附加一个Runnable的Accept事件处理者
            sk.attach(new Acceptor(sel, serverSocket));
        }

        @Override
        public void run() {
            while (!Thread.interrupted()) {
                try {
                    sel.select();
                    Set<SelectionKey> selKeySet = sel.selectedKeys();
                    Iterator<SelectionKey> it = selKeySet.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        System.out.println("key isAcceptable: " + key.isAcceptable()
                                + ", isConnectable: " + key.isConnectable()
                                + ", isReadable: " + key.isReadable()
                                + ", isWritable: " + key.isWritable()
                                + ", isValid: " + key.isValid());
                        dispatch(key);
                    }
                    selKeySet.clear();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        private void dispatch(SelectionKey selectionKey) {
            Runnable r = (Runnable) selectionKey.attachment();
            if (r != null) {
                r.run();
            }
        }
    }

    private static class Acceptor implements Runnable {
        final Selector sel;
        final ServerSocketChannel serverSocket;

        public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) {
            sel = selector;
            serverSocket = serverSocketChannel;
        }

        @Override
        public void run() {
            try {
                SocketChannel channel = serverSocket.accept();
                System.out.println("----Acceptor#run()----channel: " + (channel == null ? "null" : channel));
                if (channel != null) {
                    System.out.println("收到新的客户端连接：" + channel);
                    new Handler(sel, channel);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 客户端读写事件处理器
     */
    private static class Handler implements Runnable {
        final SocketChannel socket;
        final SelectionKey sk;
        final int MAXIN = 1024;
        ByteBuffer input = ByteBuffer.allocate(MAXIN);
        ExecutorService threadPool = CustomThreadPool.buildPool();

        public Handler(Selector selector, SocketChannel socketChannel) throws IOException {
            System.out.println("----Handler()----");
            socket = socketChannel;
            socket.configureBlocking(false);
            //获取客户端通道与Selector的绑定key
            sk = socket.register(selector, 0);
            sk.interestOps(SelectionKey.OP_READ);//读事件感兴趣
            sk.attach(this);//绑定自己为读写事件的处理器
            //selector.wakeup();//激活在select()方法上的线程，告诉它感兴趣的事件发生变化了
        }

        @Override
        public void run() {
            System.out.println("----Handler#run()----");
            try {
                if (sk.isReadable()) {
                    read();
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        void read() throws IOException {
            int len = socket.read(input);
            System.out.println("----Handler#read 读到字节数：" + len);
            if (inputIsComplete()) {
                input.flip();
                //process();//读完数据，进行业务逻辑处理
                threadPool.execute(new Processer(sk, input));
                input.clear();
            }
            if (len < 0) {
                sk.cancel();
            }
        }

        private boolean inputIsComplete() {
            return input.position() > 0;
        }
    }

    private static class Processer implements Runnable {
        final ByteBuffer input;
        final SelectionKey sk;

        public Processer(SelectionKey selKey, ByteBuffer in) {
            input = ByteBuffer.wrap(in.array(), 0, in.limit());
            sk = selKey;
        }

        @Override
        public void run() {
            //模拟业务逻辑处理
            System.out.println("收到数据：" + new String(input.array(), input.position(), input.limit()));
//            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(300));
            String msg = "Now time is : " + (new Date()).toString();
            ByteBuffer output = ByteBuffer.allocate(1024);
            output.put(msg.getBytes());
            try {
                output.flip();
//                new Sender(sk, output).run();
                new Sender(sk, output);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private static class Sender implements Runnable {
        final SocketChannel socket;
        ByteBuffer output;
        SelectionKey sk;
        final Object oldAttachment;

        public Sender(SelectionKey selKey, ByteBuffer respMsg) {
            socket = (SocketChannel) selKey.channel();
            output = ByteBuffer.wrap(respMsg.array(), 0, respMsg.limit());
            sk = selKey;
            sk.interestOps(sk.interestOps() | SelectionKey.OP_WRITE);
            oldAttachment = sk.attach(this);
            sk.selector().wakeup();//SelectionKey.interestOps与Selector.select()不在一个线程时，需要wakeup
        }

        @Override
        public void run() {
            try {
                System.out.println("----Sender#run()----");
                if (sk.isWritable()) {
                    socket.write(output);
                    if (outputIsComplete()) {
                        output.clear();
                        sk.attach(oldAttachment);
                        //只有写事件操作完成了，才取消
                        sk.interestOps(sk.interestOps() & ~SelectionKey.OP_WRITE);
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private boolean outputIsComplete() {
            return !output.hasRemaining();
        }
    }
}
